package com.bw.yk10.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;

import java.util.ArrayList;
import java.util.List;
import java.util.Locale;


public class DimHbaseSink extends RichSinkFunction<String> {
    public DimHbaseSink() {
    }
    private  Connection connection;

    // 打开hbase 连接
    @Override
    public void open(Configuration parameters) throws Exception {

        super.open(parameters);
        // 获取全局配置文件，并转为ParameterTool
        //创建一个Hbase的连接
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop-single");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        connection = ConnectionFactory.createConnection(conf);

    }

    // 写入数据
    @Override
    public void invoke(String str, Context context) throws Exception {

        JSONObject value=JSON.parseObject(str);

        //得到需要的原始数据 转化为object
        //获取到数据中的表与类型。将来能在数据中的HBASE字段的获取
        String table1 = value.getString("table");
//        String type = value.getString("op");
        String data1 = value.getString("data");
        //获取其中的data 真实的数据是其中的表数据
        JSONObject data = JSON.parseObject(data1);

        //在data 中获取到对应的主键值
        String rk = data.getString("id");

        //创建put对象，并赋rk值
        Put put = new Put(rk.getBytes());
        //将数据保存
        put.addColumn("INFO".getBytes(), "LINE".getBytes(), data1.getBytes());

        String tablename = "DIM2_"+table1.toUpperCase(Locale.ROOT);
        Table table = connection.getTable(TableName.valueOf( tablename));
        table.put(put);
        table.close();

    }

    //关闭连接
    @Override
    public void close() throws Exception {
        super.close();
        connection.close();

    }
}
